//********************************************************************
//	created:	15:10:2011   22:12
//	filename: 	asyncsocket.cpp
//	author:		tiamo
//	purpose:	async socket
//********************************************************************

#include "stdafx.h"
#include "asyncsocket.h"
#include "socketmanager.h"

#ifdef min
	#undef min
#endif

#ifdef max
	#undef max
#endif

namespace network
{
	//
	// constructor
	//
	AsyncSocket::AsyncSocket(uint32_t recvBufferSize, uint32_t sendBufferSize)
		: mSendBuffer(sendBufferSize), mRecvBuffer(recvBufferSize)
	{
		mUserData															= nullptr;
		mSocketManager														= nullptr;
		mHandle																= INVALID_SOCKET;
		mConnected															= false;
		mCanSendBuffer														= false;
		mCloseAfterSend														= false;
		memset(&mAddress, 0, sizeof(mAddress));
	}

	//
	// destructor
	//
	AsyncSocket::~AsyncSocket()
	{
		mSocketManager														= nullptr;
		close(false);
	}

	//
	// create client socket
	//
	bool AsyncSocket::createTcpClientSocket(UniString const& hostName, uint16_t port, bool nonBlock)
	{
		return connect(hostName, port, nonBlock);
	}

	//
	// create tcp server socket
	//
	void AsyncSocket::createTcpServerSocket(UniString const& listenAddress, uint16_t listenPort, bool nonBlock)
	{
		AddressHelper addressHelper;
		addressHelper.fromString(listenAddress, listenPort);
		createTcpSocket(addressHelper.getFamily());
		setReuseAddress(true);
		bindAddress(addressHelper);
		setNonBlock(nonBlock);
		listen();
	}

	//
	// attach accepted socket
	//
	void AsyncSocket::attachAcceptedSocket(SOCKET handle, AddressHelper const& address, bool nonBlock)
	{
		mHandle																= handle;
		setupAddress(nullptr, &address);
		setNonBlock(nonBlock);
	}

	//
	// create tcp socket
	//
	void AsyncSocket::createTcpSocket(int familyType, bool ignoreError)
	{
		//
		// already opened
		//
		if(mHandle != INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): already opened socket (%lu)", this, mHandle);

		//
		// create handle
		//
		mHandle																= socket(familyType, SOCK_STREAM, IPPROTO_TCP);
		if(mHandle == INVALID_SOCKET && !ignoreError)
			ThrowException(L"AsyncSocket(%p): create socket failed (%u)", this, platform::getLastErrorCode());

		LogDebug(L"AsyncSocket(%p): create socket (%lu)", this, mHandle);
	}

	//
	// bind
	//
	void AsyncSocket::bindAddress(AddressHelper const& address)
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before bind", this);

		//
		// bind
		//
		if(bind(mHandle, address.getAddress(), address.getAddressLength()) == -1)
			ThrowException(L"AsyncSocket(%p): bind (%lu) address to (%ls) failed (%u)", this, mHandle, address.toString().c_str(), platform::getLastErrorCode());

		//
		// save local address
		//
		setupAddress(&address, nullptr);
		LogDebug(L"AsyncSocket(%p): bind socket (%lu) to (%ls)", this, mHandle, address.toString().c_str());
	}

	//
	// connect to remote
	//
	bool AsyncSocket::connect(UniString const& hostName, uint16_t port, bool nonBlock)
	{
		//
		// already opened
		//
		if(mHandle != INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): already opened socket (%lu)", this, mHandle);

		//
		// get address list
		//
		UniString ipList;
		std::list<AddressHelper> addressList								= AddressHelper::getAddressList(hostName, port);
		for(auto it = addressList.begin(); it != addressList.end(); ++ it)
		{
			if(!ipList.empty())
				ipList														+= L", ";

			ipList															+= it->toString(false);
		}
		LogInfo(L"AsyncSocket(%p): resolved (%ls) to (%ls)", this, hostName.c_str(), ipList.c_str());

		//
		// connect it
		//
		for(auto it = addressList.begin(); it != addressList.end(); ++ it)
		{
			AddressHelper const& remoteAddress								= *it;
			createTcpSocket(remoteAddress.getFamily(), true);
			if(mHandle != INVALID_SOCKET)
			{
				setNonBlock(nonBlock);
				return connect(remoteAddress);
			}
		}
		ThrowException(L"AsyncSocket(%p): create socket failed (%u)", this, platform::getLastErrorCode());
	}

	//
	// connect
	//
	bool AsyncSocket::connect(AddressHelper const& peerAddress)
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before connect", this);

		//
		// connect
		//
		if(::connect(mHandle, peerAddress.getAddress(), peerAddress.getAddressLength()) == -1 && !isWouldBlockError())
		{
			LogError(L"AsyncSocket(%p): connect (%lu) to (%ls) failed (%u)", this, mHandle, peerAddress.toString().c_str(), platform::getLastErrorCode());
			onConnected(true);
			return false;
		}

		//
		// save peer address
		//
		setupAddress(nullptr, &peerAddress);
		LogDebug(L"AsyncSocket(%p): connecting socket (%lu) to (%ls)", this, mHandle, peerAddress.toString().c_str());
		return true;
	}

	//
	// listen
	//
	void AsyncSocket::listen()
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before listen", this);

		if(::listen(mHandle, 5) == -1)
			ThrowException(L"AsyncSocket(%p): listen (%lu) at (%ls) failed (%u)", this, mHandle, getAddress(true).toString(true).c_str(), platform::getLastErrorCode());;

		LogDebug(L"AsyncSocket(%p): listen socket (%lu) at (%ls)", this, mHandle, getAddress(true).toString(true).c_str());
	}

	//
	// accept
	//
	SOCKET AsyncSocket::accept(AddressHelper& address)
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before accept", this);

		//
		// accept
		//
		uint8_t addressBuffer[128]											= {0};
		sockaddr* acceptedAddress											= reinterpret_cast<sockaddr*>(addressBuffer);
		socklen_t addressLength												= sizeof(addressBuffer);
		SOCKET newSocket													= ::accept(mHandle, acceptedAddress, &addressLength);
		if(newSocket != INVALID_SOCKET)
		{
			address.attach(acceptedAddress, addressLength);
			LogDebug(L"AsyncSocket(%p): accept new socket (%lu) at (%ls)", this, newSocket, address.toString().c_str());
		}

		return newSocket;
	}

	//
	// send buffer
	//
	uint32_t AsyncSocket::sendBuffer(void const* inputBuffer, uint32_t inputLength)
	{
		//
		// check param
		//
		if(!inputLength || !inputBuffer)
			ThrowException(L"AsyncSocket(%p): invalid param (%p,%u) for send to socket (%lu)", this, inputBuffer, inputLength, mHandle);

		//
		// append to send buffer
		//
		{
			utils::Locker locker(mSendBufferLock);
			mSendBuffer.appendData(inputBuffer, inputLength);
		}

		if(mSocketManager)
			mSocketManager->addFlushSocket(this);

		return inputLength;
	}

	//
	// copy send buffer
	//
	uint32_t AsyncSocket::readSendBuffer(void* outputBuffer, uint32_t bufferLength)
	{
		utils::Locker locker(mSendBufferLock);
		bufferLength														= std::min(bufferLength, mSendBuffer.getLength());
		if(outputBuffer)
			memcpy(outputBuffer, mSendBuffer.getBuffer(), bufferLength);

		return bufferLength;
	}

	//
	// consume send buffer
	//
	void AsyncSocket::consumeSendBuffer(uint32_t bufferLength)
	{
		utils::Locker locker(mSendBufferLock);
		if(bufferLength > mSendBuffer.getLength())
			ThrowException(L"AsyncSocket(%p): invalid buffer length(%u > %u)", this, bufferLength, mSendBuffer.getLength());

		mSendBuffer.consumeData(bufferLength);
	}

	//
	// set non-block
	//
	void AsyncSocket::setNonBlock(bool nonBlock)
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"SocketFile(%p): should create socket before set non-block", this);

	#ifdef WIN32
		unsigned long optValue												= nonBlock ? 1 : 0;
		if(ioctlsocket(mHandle, FIONBIO, &optValue) == -1)
			ThrowException(L"AsyncSocket(%p): set block mode (%lu) for socket (%lu) failed (%u)", this, optValue, mHandle, platform::getLastErrorCode());
	#else
		int theFlags															= fcntl(mHandle, F_GETFL, 0);
		if(theFlags == -1)
			ThrowException(L"AsyncSocket(%p): set block mode (%lu) for socket (%lu) failed(get %u)", this, nonBlock, mHandle, platform::getLastErrorCode());

		if(nonBlock)
			theFlags														|= O_NONBLOCK;
		else
			theFlags														&= ~O_NONBLOCK;

		if(fcntl(mHandle, F_SETFL, theFlags) == -1)
			ThrowException(L"AsyncSocket(%p): set block mode (%lu) for socket (%lu) failed(set %u)", this, nonBlock, mHandle, platform::getLastErrorCode());
	#endif

		LogDebug(L"AsyncSocket(%p): set %lsblock mode for socket (%lu)", this, nonBlock ? L"non" : L"", mHandle);
	}

	//
	// set reuse-address
	//
	void AsyncSocket::setReuseAddress(bool reuse)
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before set reuseaddr", this);

		int reuseAddress													= reuse ? 1 : 0;
		if(setsockopt(mHandle, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast<char const*>(&reuseAddress), sizeof(reuseAddress)) == -1)
			ThrowException(L"AsyncSocket(%p): set reuse address(%d) for socket (%lu) failed (%u)", this, reuseAddress, mHandle, platform::getLastErrorCode());

		LogDebug(L"AsyncSocket(%p): set reuse(%d) address for socket (%lu)", this, reuse, mHandle);
	}

	//
	// get socket error
	//
	int AsyncSocket::getSocketError()
	{
		//
		// must be opened
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before get error code", this);

		int errorCode														= 0;
		socklen_t length													= sizeof(errorCode);
		if(getsockopt(mHandle, SOL_SOCKET, SO_ERROR, reinterpret_cast<char*>(&errorCode), &length) == -1)
			ThrowException(L"AsyncSocket(%p): get error code from socket (%lu) failed (%u)", this, mHandle, platform::getLastErrorCode());

		return errorCode;
	}

	//
	// is would block error
	//
	bool AsyncSocket::isWouldBlockError() const
	{
	#ifdef WIN32
		return platform::getLastErrorCode() == WSAEWOULDBLOCK;
	#else
		return platform::getLastErrorCode() == EWOULDBLOCK || platform::getLastErrorCode() == EAGAIN || platform::getLastErrorCode() == EINTR || platform::getLastErrorCode() == EINPROGRESS;
	#endif
	}

	//
	// setup address
	//
	void AsyncSocket::setupAddress(AddressHelper const* localAddress, AddressHelper const* peerAddress)
	{
		if(!localAddress)
		{
			uint8_t addressBuffer[128]										= {0};
			socklen_t length												= sizeof(addressBuffer);
			sockaddr* address												= reinterpret_cast<sockaddr*>(addressBuffer);
			if(!getsockname(mHandle, address, &length))
				mAddress[0].attach(address, length);
		}
		else
		{
			mAddress[0]														= *localAddress;
		}

		if(!peerAddress)
		{
			uint8_t addressBuffer[128]										= {0};
			socklen_t length												= sizeof(addressBuffer);
			sockaddr* address												= reinterpret_cast<sockaddr*>(addressBuffer);
			if(!getpeername(mHandle, address, &length))
				mAddress[1].attach(address, length);
		}
		else
		{
			mAddress[1]														= *peerAddress;
		}
	}

	//
	// network event
	//
	void AsyncSocket::onNetworkEvent(bool readEvent, bool writeEvent, bool errorEvent)
	{
		//
		// catch exception
		//
		try
		{
			//
			// read/write event
			//
			if(readEvent || writeEvent)
			{
				if(!mConnected)
				{
					mConnected												= true;
					LogDebug(L"AsyncSocket(%p): socket (%lu) connected", this, mHandle);
					onConnected(false);
				}
			
				if(readEvent)
					readFromSocket();

				if(writeEvent)
					writeDataToSocket();
			}

			//
			// check error event
			//
			if(errorEvent)
				onError(nullptr);
		}
		catch(std::exception& theException)
		{
			onError(&theException);
		}
	}

	//
	// flush write buffer
	//
	void AsyncSocket::flushWriteBuffer()
	{
		try
		{
			if(mConnected && mCanSendBuffer)
				writeDataToSocket();
		}
		catch(std::exception& theException)
		{
			onError(&theException);
		}
	}

	//
	// send buffer
	//
	uint32_t AsyncSocket::writeDataToSocket()
	{
		//
		// already closed
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before send", this);

		uint32_t totalLength												= 0;
		mCanSendBuffer														= true;
		while(1)
		{
			//
			// read send buffer
			//
			char inputBuffer[1024]											= {0};
			uint32_t inputLength											= readSendBuffer(inputBuffer, sizeof(inputBuffer));
			if(!inputLength)
				break;

			//
			// send it
			//
			int length														= realWrite(inputBuffer, inputLength);

			//
			// closed
			//
			if(length == 0)
				ThrowException(L"AsyncSocket(%p): socket (%lu) closed (%u)", this, mHandle, platform::getLastErrorCode());

			//
			// send error
			//
			if(length < 0 && !isWouldBlockError())
				ThrowException(L"AsyncSocket(%p): send (%p,%u) for socket (%lu) failed (%u)", this, inputBuffer, inputLength, mHandle, platform::getLastErrorCode());

			//
			// consume data
			//
			if(length > 0)
			{
				totalLength													+= length;
				consumeSendBuffer(length);
			}

			//
			// buffer full
			//
			if(length < static_cast<int>(inputLength))
				mCanSendBuffer												= false;

			//
			// done
			//
			if(!mCanSendBuffer)
				break;
		}

		if(mCanSendBuffer && mCloseAfterSend)
			mSocketManager->closeSocket(this, true);

		return totalLength;
	}

	//
	// read from socket
	//
	uint32_t AsyncSocket::readFromSocket()
	{
		//
		// already closed
		//
		if(mHandle == INVALID_SOCKET)
			ThrowException(L"AsyncSocket(%p): should create socket before recv", this);

		//
		// loop read
		//
		uint32_t totalReadLength											= 0;
		while(1)
		{
			//
			// read 1KB to temp buffer
			//
			char tempBuffer[1024]											= {0};
			int length														= realRead(tempBuffer, sizeof(tempBuffer));

			//
			// closed
			//
			if(length == 0)
				ThrowException(L"AsyncSocket(%p): socket (%lu) closed (%u)", this, mHandle, platform::getLastErrorCode());

			//
			// recv error
			//
			if(length < 0 && !isWouldBlockError())
				ThrowException(L"AsyncSocket(%p): recv for socket (%lu) failed (%u)", this, mHandle, platform::getLastErrorCode());

			//
			// append temp buffer to recv buffer
			//
			if(length > 0)
			{
				mRecvBuffer.appendData(tempBuffer, length);
				totalReadLength												+= length;
			}

			//
			// done
			//
			if((length < 0 && isWouldBlockError()) || length < sizeof(tempBuffer))
				break;
		}

		//
		// build packet
		//
		if(totalReadLength)
			while(processReceivedData()){}

		return totalReadLength;
	}

	//
	// error event
	//
	void AsyncSocket::onError(std::exception* theException)
	{
		if(theException)
		{
			LogError(L"AsyncSocket(%p): socket(%lu) met an exception: %ls", this, mHandle, utils::Exception::getMessage(*theException).c_str());
		}
		else
		{
			LogError(L"AsyncSocket(%p): socket(%lu) met a network error (%u)", this, mHandle, getSocketError());
		}

		mSocketManager->closeSocket(this, true);
	}

	//
	// close
	//
	void AsyncSocket::close(bool errorState)
	{
		if(mHandle != INVALID_SOCKET)
		{
		#ifdef WIN32
			closesocket(mHandle);
		#else
			::close(mHandle);
		#endif
			mHandle															= INVALID_SOCKET;

			onSocketClosed(errorState);
		}
	}

	//
	// connect
	//
	void AsyncSocket::onConnected(bool errorState)
	{
		if(mSocketManager)
			mSocketManager->pushConnectedPacket(this, errorState);

		if(errorState)
			close(errorState);
	}

	//
	// closed
	//
	void AsyncSocket::onSocketClosed(bool errorState)
	{
		if(mSocketManager)
			mSocketManager->pushClosedPacket(this, errorState);
	}

	//
	// process recieved data
	//
	bool AsyncSocket::processReceivedData()
	{
		if(!mDecodeDataCallback || !mRecvBuffer.getLength())
			return false;

		uint32_t usedLength													= 0;
		utils::SmartPointer<BasePacket> thePacket							= mDecodeDataCallback(this, mRecvBuffer.getBuffer(), mRecvBuffer.getLength(), usedLength);
		if(usedLength)
			mRecvBuffer.consumeData(usedLength);

		if(thePacket)
			mSocketManager->pushDataPacket(this, std::move(thePacket));

		return !!usedLength;
	}
}
